﻿using Basic.EventBus.Attributes;
using Basic.EventBus.Eventbus;
using Confluent.Kafka;

namespace Basic.EventBus.EventBusKafka;

public class KafkaProducer : IEventPublisher
{
    private KafkaBusBrokerConfig BrokerConfig;
    private IProducer<string, byte[]> Producer;

    public KafkaProducer(KafkaBusBrokerConfig options)
    {
        BrokerConfig = options;

        var conf = new ProducerConfig { BootstrapServers = BrokerConfig.Servers ?? ($"{BrokerConfig.Host}:{BrokerConfig.Port}") };
        var builder = new ProducerBuilder<string, byte[]>(conf);
        Producer = builder.Build();
    }

    public async Task<bool> PublishAsync(byte[] data, EventTopicAttribute attribute)
    {
        var message = new Message<string, byte[]> { Key = attribute.Key, Value = data.ToArray() };
        var resp = await Producer.ProduceAsync(attribute.Topic, message);
        return resp.Status > 0;
    }

    public async Task<bool> PublishAsync(IEventData eventData)
    {
        var attribute = EventTopicAttribute.GetCurrentAttribute(eventData.GetType());
        if (attribute == null)
            return false;
        return await PublishAsync(eventData.Serialize(), attribute);
    }

    public async Task<int> PublishAsync(List<IEventData> eventDatas)
    {
        var attribute = EventTopicAttribute.GetCurrentAttribute(eventDatas[0].GetType());
        if (attribute == null) return 0;

        foreach (var eventData in eventDatas)
            await PublishAsync(eventData.Serialize(), attribute);

        return eventDatas.Count;
    }

    #region Dispose
    private bool disposedValue;

    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                Producer.Dispose();
            }
            // TODO: 将大型字段设置为 null
            Producer = null;
            disposedValue = true;
        }
    }

    public void Dispose()
    {
        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
    #endregion

}